package com.sui.bigdata.sml.web.service.impl;

import com.feidee.fdcommon.azkaban.model.SparkJobParam;
import com.sui.bigdata.sml.web.dto.*;
import com.sui.bigdata.sml.web.exception.EntityNotFoundException;
import com.sui.bigdata.sml.web.misc.azkaban.AzkabanHelper;
import com.sui.bigdata.sml.web.repository.mapper.*;
import com.sui.bigdata.sml.web.repository.model.AzProject;
import com.sui.bigdata.sml.web.repository.model.Connection;
import com.sui.bigdata.sml.web.repository.model.Node;
import com.sui.bigdata.sml.web.repository.model.NodeGroup;
import com.sui.bigdata.sml.web.service.AzkabanService;
import com.sui.bigdata.sml.web.service.NodeService;
import com.sui.bigdata.sml.web.util.Constants;
import com.sui.bigdata.sml.web.vo.RunNodeVO;
import lombok.extern.slf4j.Slf4j;
import ma.glasnost.orika.MapperFacade;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @author songhaicheng
 * @date 2019/11/7 10:40
 * @description
 * @reviewer
 */
@Service
@DependsOn("disconfConfig")
@Slf4j
public class NodeServiceImpl implements NodeService {

    @Autowired
    private NodeMapper mapper;
    @Autowired
    private ProjectMapper projectMapper;
    @Autowired
    private ExperimentMapper experimentMapper;
    @Autowired
    private ComponentMapper componentMapper;
    @Autowired
    private NodeGroupMapper groupMapper;
    @Autowired
    private AzkabanService azkabanService;
    @Autowired
    private MapperFacade mapperFacade;

    @Override
    public Long add(NodeDTO dto) {
        Node node = mapperFacade.map(dto, Node.class);
        mapper.insert(node);
        return node.getId();
    }

    @Override
    public NodeDTO get(Long id) {
        Node node = mapper.get(id);
        if (node == null) {
            throw new EntityNotFoundException(Node.class, id);
        }
        return mapperFacade.map(node, NodeDTO.class);
    }

    @Override
    public Boolean delete(Long id) {
        get(id);
        return mapper.delete(id) == 1;
    }

    @Override
    public List<NodeDTO> listByExperiment(ExperimentDTO experimentDto) {
        return mapperFacade.mapAsList(mapper.listByExperimentId(experimentDto.getId()), NodeDTO.class);
    }

    @Override
    public List<NodeDTO> listByComponent(ComponentDTO componentDto) {
        return mapperFacade.mapAsList(mapper.listByComponentId(componentDto.getId()), NodeDTO.class);
    }

    @Transactional
    @Override
    public Boolean updateBatch(List<NodeDTO> dtos) {
        if (dtos.size() == 0) {
            return true;
        }
        List<Node> nodes = dtos
                .parallelStream()
                .map(dto -> mapperFacade.map(dto, Node.class)).collect(Collectors.toList());
        // be tolerant with the batch updating number's inconsistent
        return mapper.updateBatch(nodes) > 0;
    }

    @Override
    public List<ConnectionDTO> listConnectionsByNode(NodeDTO dto) {
        return mapperFacade.mapAsList(mapper.listConnectionsByNodeId(dto.getId()), ConnectionDTO.class);
    }

    @Override
    public List<ConnectionDTO> listAllConnectionsBetweenNodes(List<NodeDTO> dtos) {
        if (dtos == null || dtos.isEmpty()) {
            return new ArrayList<>();
        }
        List<Long> nodeIds = dtos.parallelStream().map(NodeDTO::getId).collect(Collectors.toList());
        return mapperFacade.mapAsList(mapper.listConnectionsBetweenNodeIds(nodeIds), ConnectionDTO.class);
    }

    @Override
    public Long addConnection(ConnectionDTO dto) {
        Connection connection = mapperFacade.map(dto, Connection.class);
        mapper.insertConnection(connection);
        return connection.getId();
    }

    @Override
    public Boolean deleteConnections(List<Long> connectionIds) {
        return mapper.deleteConnections(connectionIds) == connectionIds.size();
    }

    @Transactional
    @Override
    public RunNodeVO run(NodeDTO entry, String type) {
        // 解析出会跑的节点
        List<AzkabanHelper.Node> runningNodes = new ArrayList<>();
        ExperimentDTO experiment = mapperFacade.map(experimentMapper.get(entry.getExperimentId()), ExperimentDTO.class);
        ProjectDTO project = mapperFacade.map(projectMapper.get(experiment.getProjectId()), ProjectDTO.class);
        AzProject azProject = AzkabanHelper.initProject();
        azProject.setDescription("算法前台前端界面提交的任务（用户：" + experiment.getUserId() +
                "，项目：" + project.getName() + "，实验：" + experiment.getName() + "）");
        // 加个 endJob，以实现 az 的任务并行化
        SparkJobParam endJob = null;
        // 选择运行方式
        if (AzkabanHelper.RunningType.getRunningType(type) == AzkabanHelper.RunningType.SINGLE) {
            if (entry.getStatus() == 0) {
                throw new RuntimeException("组件【" + entry.getName() + "】必填参数未设置完整");
            }
            azProject.setDescription("算法前台前端界面提交的单节点任务（用户：" + experiment.getUserId() +
                    "，项目：" + project.getName() + "，实验：" + experiment.getName() +
                    "，节点：" + entry.getName() + "）");
            runningNodes.add(new AzkabanHelper.Node(entry));
        } else {
            List<NodeDTO> allNodes = listByExperiment(experiment);
            List<ConnectionDTO> connections = listAllConnectionsBetweenNodes(allNodes);
            Map<Long, NodeDTO> id2Node = new HashMap<>();
            allNodes.parallelStream().forEach(n -> id2Node.put(n.getId(), n));
            // 选择多节点运行方式
            switch (AzkabanHelper.RunningType.getRunningType(type)) {
                case FROM:
                    runningNodes = AzkabanHelper.initGraph(id2Node, connections, entry);
                    break;
                case TO:
                    // 将连线的头尾部转换过来
                    connections.parallelStream().forEach(conn -> {
                        Long hid = conn.getHeadNid();
                        conn.setHeadNid(conn.getTailNid());
                        conn.setTailNid(hid);
                    });
                    runningNodes = AzkabanHelper.initGraph(id2Node, connections, entry);
                    // 解析出的节点头尾再转回来
                    runningNodes.parallelStream().forEach(n -> {
                        List<AzkabanHelper.Node> parents = n.parents;
                        n.parents = n.children;
                        n.children = parents;
                    });
                    break;
                default:
                    // ALL
                    runningNodes = AzkabanHelper.initGraph(id2Node, connections, null);
                    break;
            }
            endJob = AzkabanHelper.appendEndJob(experiment.getName(), runningNodes);
        }
        if (runningNodes.size() == 0) {
            throw new RuntimeException("无可运行节点");
        }
        // 检测即将运行的节点是否能跑（满足下列条件）
        //  1、必填参数设置完整
        List<AzkabanHelper.Node> lacks = runningNodes
                .parallelStream()
                .filter(n -> n.val.getStatus() == 0)
                .collect(Collectors.toList());
        if (lacks.size() > 0) {
            throw new RuntimeException("节点【" + lacks.get(0).val.getName() + "】必填参数未设置完整");
        }
        //  2、之前未被加入“等待”、“运行中”的节点组中
        List<Long> ongoingIds = groupMapper.listOngoingsByExperimentId(experiment.getId())
                .parallelStream()
                .flatMap(g -> Arrays.stream(g.getNodeIds().split(",")).map(Long::valueOf))
                .collect(Collectors.toList());
        List<AzkabanHelper.Node> stillRunning = runningNodes
                .parallelStream()
                .filter(n -> ongoingIds.contains(n.val.getId()))
                .collect(Collectors.toList());
        if (stillRunning.size() > 0) {
            throw new RuntimeException("节点【" + stillRunning.get(0).val.getName() + "】仍处于某一未完成的运行队列中");
        }
        // 3、将新的要跑节点加入运行队列中
        NodeGroup group = new NodeGroup();
        group.setExperimentId(experiment.getId());
        group.setUserId(Constants.DEFAULT_USER_ID);
        group.setNodeIds(String.join(",", runningNodes.stream().map(n -> n.val.getId().toString()).toArray(String[]::new)));
        group.setAzProjectName(azProject.getName());
        groupMapper.insert(group);

        // 初始化要跑的节点
        AzkabanHelper.initNodes(runningNodes);
        List<SparkJobParam> jobs = runningNodes.parallelStream().map(rn ->
                AzkabanHelper.createSparkJob(
                        rn,
                        componentMapper.get(rn.val.getComponentId()).getAlgClassName(),
                        group.getId()
                )
        ).collect(Collectors.toList());
        if (endJob != null) {
            jobs.add(endJob);
        }
        azProject.setJobs(jobs);

        azkabanService.createProject(azProject);

        // 更新的节点信息
        List<NodeDTO> nodes = runningNodes.parallelStream().map(n -> n.val).collect(Collectors.toList());
        updateBatch(nodes);

        log.info("提交 Azkaban 任务成功：{}", azProject.toString());

        return new RunNodeVO()
                .setName(azProject.getName())
                .setNodes(nodes)
                .setNewGroup(mapperFacade.map(group, NodeGroupDTO.class));
    }

    @Override
    public NodeResultDTO getResultByNode(NodeDTO dto) {
        return mapperFacade.map(mapper.getResultById(dto.getId()), NodeResultDTO.class);
    }

    @Override
    public List<NodeResultDTO> getResultHistoriesByNode(NodeDTO dto) {
        return mapperFacade.mapAsList(mapper.getResultHistoriesById(dto.getId()), NodeResultDTO.class);
    }

}
